import org.apache.kafka.clients.consumer.KafkaConsumer
import org.apache.kafka.common.serialization.ByteArrayDeserializer
import org.apache.spark.{ SparkConf, TaskContext }
import org.apache.spark.streaming.{Milliseconds, StreamingContext}
import org.apache.spark.streaming.kafka010.{ KafkaUtils, HasOffsetRanges, OffsetRange }
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import com.typesafe.config.ConfigFactory


object Throughout {

  /**
    *提升吞吐量可以设置：一组topic进行消费 1各类日志---N个topic---N个分区，提升吞吐量
    * @param args
    */
  def main(args: Array[String]): Unit = {

    val conf = ConfigFactory.load
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> conf.getString("kafka.brokers"),
      "key.deserializer" -> classOf[ByteArrayDeserializer],
      "value.deserializer" -> classOf[ByteArrayDeserializer],
      "group.id" -> conf.getString("kafka.groupid"),
      "auto.offset.reset" -> "latest")
    val topics = conf.getString("kafka.topics").split(",")
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName(s"${this.getClass.getSimpleName}")
    val ssc = new StreamingContext(sparkConf, Milliseconds(conf.getLong("batchDurationMs")))

    val stream = KafkaUtils.createDirectStream(
      ssc,
      PreferConsistent,
      Subscribe[Array[Byte],Array[Byte]](topics,kafkaParams)
    )
    stream.foreachRDD{rdd=>
      println(rdd.map(_.value().size.toLong).fold(0)(_+_))
    }
    ssc.start()
    ssc.awaitTermination()
  }
}
